package com.gitee.cnlongs.cnlongmq.server.handler;

import java.util.Date;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.gitee.cnlongdb.basic.entity.datasync.CnLongDataSyncRequest;
import com.gitee.cnlongdb.basic.entity.datasync.CnLongDataSyncResponse;
import com.gitee.cnlongdb.basic.entity.query.LongDbQuery;
import com.gitee.cnlongdb.basic.entity.query.LongDbWhere;
import com.gitee.cnlongdb.basic.utils.RandomUtil;
import com.gitee.cnlongdb.server.datasync.DataSyncConnectMessage;
import com.gitee.cnlongdb.server.transaction.actuator.LongdbTransactionActuator;
import com.gitee.cnlongs.cnlongmq.basic.constant.CnLongMqConstant;
import com.gitee.cnlongs.cnlongmq.basic.entitiy.request.CnLongMessageQueue;
import com.gitee.cnlongs.cnlongmq.basic.entitiy.request.CnLongMqConfiger;
import com.gitee.cnlongs.cnlongmq.basic.entitiy.request.CnLongMqMessage;
import com.gitee.cnlongs.cnlongmq.basic.entitiy.response.CnLongMqResponse;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqDataSync;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqExceptionCode;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqMode;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqRole;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqStatus;
import com.gitee.cnlongs.cnlongmq.server.LongMqServer;
import com.gitee.cnlongs.cnlongmq.server.operator.CnLongMqExecuter;
import com.gitee.cnlongs.cnlongmq.server.threads.datasync.CnLongMqExchangeThread;
import com.gitee.cnlongs.cnlongmq.server.threads.message.HandlerLongMqMessage;

/**
 * 系统名称：LongMqRequest
 * 模块名称：
 * 模块描述：mq消息发送请求类
 * 功能列表：
 * 模块作者：zouyong
 * 开发时间：2019年4月12日 下午6:09:31
 * 模块路径：com.gitee.cnlongs.cnlongmq.server.handler.LongMqRequest
 */
public class LongMqRequest {

	private static final Logger logger = (Logger) LoggerFactory.getLogger(CnLongMqExchangeThread.class);
	
	/**
	 * 功能描述：发起MQ消息发送请求
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月12日 下午5:58:50
	 × 返回数据：String
	 */
	public static CnLongMqResponse launchSendMessage(CnLongMqMessage mqMessage) throws Exception {
		try {
			CnLongDataSyncRequest dataSync = new CnLongDataSyncRequest();
			dataSync.setMessageType(mqMessage.getReqType());
			dataSync.setClassPath(LongMqRequest.class.getName());
			dataSync.setMethodName(CnLongMqDataSync.receiveLongMqMessage.name());
			dataSync.setParamClas(new String[]{ mqMessage.getClass().getName() });
			dataSync.setParamObject(new String[]{ JSON.toJSONString(mqMessage) });
			dataSync.setOperThreadId(RandomUtil.getCharAndNumr(20));
			
			// 是否发送mq消息
			boolean sendLongMqMessage = CnLongMqDataSync.sendLongMqMessage.name().equals(mqMessage.getReqType());
			// 是否执行ack确认
			boolean noticeAckResult = CnLongMqDataSync.noticeAckResult.name().equals(mqMessage.getReqType());
			// 输出内容
			String opermsg = sendLongMqMessage ? "MQ消息发送" : "MQ消息ack确认";
			logger.debug(opermsg + "请求：" + JSON.toJSONString(mqMessage));
			
			CnLongDataSyncResponse response = null;
			// 发送MQ消息
			if(sendLongMqMessage) {
				response = DataSyncConnectMessage
						.noticeDataSyncSend(mqMessage.getCustomerServerIp(), mqMessage.getCustomerPort(), dataSync);
			}
			// 返回ACK确认
			else if(noticeAckResult) {
				response = DataSyncConnectMessage
						.noticeDataSyncSend(mqMessage.getSendServerIp(), mqMessage.getSendServerPort(), dataSync);
			}
			if(null == response) {
				throw new RuntimeException(opermsg + "：请求类型为空或发起MQ请求失败！");
			}
			logger.debug(opermsg + "结果：" + JSON.toJSONString(response));
			
			String jsonData = response.getJsonData();
			if(StringUtils.isNotBlank(jsonData)) {
				// 获取返回的消息数据
				CnLongMqMessage resultMqMsg = JSON.parseObject(jsonData, CnLongMqMessage.class);
				// 发送MQ消息
				if(sendLongMqMessage) {
					mqMessage = setLongMqMessageSendInfo(mqMessage, resultMqMsg);
					mqMessage.setReturnMessage(resultMqMsg.getReturnMessage());
				}
				// 返回ACK确认
				else if(noticeAckResult) {
					mqMessage.setNoticeSenderStatus(true); // 通知发送方成功
					mqMessage.setAckConfirm(true); // 消息ack确认成功
					mqMessage.setLongMqRole(CnLongMqRole.customer.name());
				}
				mqMessage.setErrorCode(resultMqMsg.getErrorCode());
				mqMessage.setReceiveErrMsg(resultMqMsg.getReceiveErrMsg());
				
				// 更新已发送的mq消息
				LongDbQuery query = new LongDbQuery();
				query.setTableName(mqMessage.getLongMqKey());
				query.setClassz(CnLongMqMessage.class);
				query.setPageSize(1);
				query.addLongDbWhere(LongDbWhere.where(CnLongMqDataSync.longMqId.name()).is(mqMessage.getLongMqId()));
				boolean updStatus = LongMqServer.getExecuter().getLongDbServer().update(query, mqMessage);
				logger.debug("更新状态：" + updStatus);
				
				// ACK确认完成后，将【处理中的消息数据】删除，写入到【处理成功】中
				if(noticeAckResult && StringUtils.isBlank(resultMqMsg.getErrorCode())) {
					processTransferMqMessage(mqMessage, CnLongMqStatus.success.name());
				}
				// 消费方保存消息失败 消费方没有找到队列基本数据
				else if(CnLongMqExceptionCode.SAVE_MQ_MESSAGE_FAIL_CUSTOMER.name().equals(resultMqMsg.getErrorCode())
						|| CnLongMqExceptionCode.NOT_FOUND_SEQ_BASE_CUSTOMER.name().equals(resultMqMsg.getErrorCode())) {
					return null; // 消费方异常后结束，设置发送失败，以便重新发送
				}
				// 消费方处理消息失败
				else if(CnLongMqExceptionCode.MESSAGE_HANDLER_FAIL_CUSTOMER.name().equals(resultMqMsg.getErrorCode())) {
					if(null == resultMqMsg.getReturnMessage() || resultMqMsg.getReturnMessage().getSendAgain()) {
						return null; // 消费方对消息处理失败，需要重发
					} else {
						// 处理失败，不需要重发，则直接废弃
						LongMqRequest.processTransferMqMessage(mqMessage, CnLongMqStatus.failure.name());
					}
				}
			}
		}
		catch(Exception e) {
			String error = "MQ消息发送异常：" + e.getMessage();
			logger.error(error);
			throw new RuntimeException(error);
		}
		return mqMessage.getReturnMessage();
	}
	
	/**
	 * 功能描述：记录发送信息
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年5月5日 下午5:34:55
	 × 返回数据：CnLongMqMessage
	 */
	private static CnLongMqMessage setLongMqMessageSendInfo(CnLongMqMessage mqMessage, CnLongMqMessage resultMqMsg) {
		boolean customerStatus = true;
		if(null != resultMqMsg) {
			customerStatus = null != resultMqMsg.getCustomerStatus() ? resultMqMsg.getCustomerStatus() : false;
			mqMessage.setCustomerDate(resultMqMsg.getCustomerDate()); // 消费端接收时间
		} else {
			mqMessage.setCustomerDate(new Date());		// 消费端接收时间
		}
		mqMessage.setSendStatus(true); 					// 消息发送状态：成功
		mqMessage.setCustomerStatus(customerStatus); 	// 消费端接收状态
		
		// 保存已发送消息的客户端信息和状态
		String key = mqMessage.getCustomerServerIp() + CnLongMqConstant.longmqSeparateF + mqMessage.getCustomerPort();
		mqMessage.getSendCustomerMap().put(key, customerStatus);
		
		return mqMessage;
	}

	/**
	 * 功能描述：接收MQ消息 / 执行ACK确认
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月15日 下午4:56:36
	 × 返回数据：String
	 */
	public static String receiveLongMqMessage(CnLongMqMessage mqMessage) {
		// 发送MQ消息
		if(CnLongMqDataSync.sendLongMqMessage.name().equals(mqMessage.getReqType())) 
		{
			mqMessage = executeLongMqMessage(mqMessage);
		}
		// 返回ACK确认：暂时只需要更新消息数据即可
		else if(CnLongMqDataSync.noticeAckResult.name().equals(mqMessage.getReqType())) {
			try {
				LongDbQuery query = new LongDbQuery();
				query.setTableName(mqMessage.getLongMqKey());
				query.setClassz(CnLongMqMessage.class);
				query.setPageSize(1);
				query.addLongDbWhere(LongDbWhere.where(CnLongMqDataSync.longMqId.name()).is(mqMessage.getLongMqId()));
				mqMessage.setLongMqRole(CnLongMqRole.provider.name());
				mqMessage.setNoticeSenderStatus(true); // 通知发送方成功
				mqMessage.setAckConfirm(true); // 消息ack确认成功
				LongMqServer.getExecuter().getLongDbServer().update(query, mqMessage);
			} catch(Exception e) {
				mqMessage.setErrorCode(CnLongMqExceptionCode.UPDA_MQ_MESSAGE_FAIL_PROVIDER.name());
				mqMessage.setReceiveErrMsg(e.getMessage());
				logger.error("接收ACK确认失败：" + e.getMessage());
			}
			
			try {
				// ACK确认完成后，将【处理中的消息数据】删除，写入到【处理成功】中
				if(StringUtils.isBlank(mqMessage.getErrorCode())) {
					processTransferMqMessage(mqMessage, CnLongMqStatus.success.name());
				}
				// 消费方处理消息失败
				else if(CnLongMqExceptionCode.MESSAGE_HANDLER_FAIL_CUSTOMER.name().equals(mqMessage.getErrorCode())) {
					if(null == mqMessage.getReturnMessage() || !mqMessage.getReturnMessage().getSendAgain()) {
						// 处理失败，不需要重发，则直接废弃
						LongMqRequest.processTransferMqMessage(mqMessage, CnLongMqStatus.failure.name());
					}
				}
				mqMessage.setErrorCode(null);
			} catch(Exception e) {
				mqMessage.setErrorCode(CnLongMqExceptionCode.ACK_SUCC_TRANSFER_FAIL_CUSTOMER.name());
				mqMessage.setReceiveErrMsg(e.getMessage());
				logger.error("接收ACK确认完成，转移消息数据失败：" + e.getMessage());
			}
		}
		
		return JSON.toJSONString(mqMessage);
	}

	/**
	 * 功能描述：处理接收到的MQ消息
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月16日 下午5:11:19
	 × 返回数据：CnLongMqMessageEntity
	 */
	private static CnLongMqMessage executeLongMqMessage(CnLongMqMessage mqMessage) {

		// 消息接收成功
		mqMessage.setCustomerStatus(true);
		// 消息接收时间
		mqMessage.setCustomerDate(new Date());
		
		try {
			try {
				mqMessage = setLongMqMessageSendInfo(mqMessage, null);
				mqMessage.setLongMqRole(CnLongMqRole.customer.name());
				// 将消息存储到磁盘
				LongMqServer.getExecuter().saveMqMessage(mqMessage);
			} catch(Exception e) {
				mqMessage.setErrorCode(CnLongMqExceptionCode.SAVE_MQ_MESSAGE_FAIL_CUSTOMER.name());
				throw new RuntimeException(e);
			}

			CnLongMqConfiger mqConfig = LongMqServer.getExecuter().getLongMqConfig();
			CnLongMessageQueue sequeue = CnLongMqExecuter.getSequeueByKey(mqMessage.getLongMqKey(), mqConfig);
			if(null == sequeue) {
				mqMessage.setErrorCode(CnLongMqExceptionCode.NOT_FOUND_SEQ_BASE_CUSTOMER.name());
				throw new RuntimeException(mqMessage.getErrorCode());
			}

			// 组装消息内容对象
			CnLongMqResponse message = new CnLongMqResponse();
			message.setMessageContent(mqMessage.getLongMqText());
			message.setLongMqKey(mqMessage.getLongMqKey());
			
			// 消息异步执行
			if(CnLongMqMode.async.name().equals(sequeue.getMode())) {
				new HandlerLongMqMessage(mqMessage, message).start();
			}
			// 消息同步执行
			else if(CnLongMqMode.sync.name().equals(sequeue.getMode())) {
				message = handlerMqMessage(mqMessage, message);
			}
			mqMessage.setReturnMessage(message);
		}
		catch(Exception e) {
			mqMessage.setCustomerStatus(false);
			mqMessage.setReceiveErrMsg(e.getMessage());
			logger.error("错误码：{}，错误原因：{}", mqMessage.getErrorCode(), e.getMessage());
		}

		return mqMessage;
	}

	/**
	 * 功能描述：接收消息后进行业务处理
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月16日 下午3:10:11
	 × 返回数据：CnLongMqMessage
	 */
	public static CnLongMqResponse handlerMqMessage(CnLongMqMessage mqMessage, CnLongMqResponse message) throws Exception {
		
		try {
			// 调用消费端处理消息
			Object operator = Class.forName(mqMessage.getClassPath()).newInstance();
			// 反射调用处理端对象方法
			operator.getClass()
				.getMethod(CnLongMqDataSync.processLongMqMessage.name(), new Class[] { CnLongMqResponse.class })
				.invoke(operator, new Object[]{ message });
		}
		catch(Exception e) {
			mqMessage.setErrorCode(CnLongMqExceptionCode.MESSAGE_HANDLER_FAIL_CUSTOMER.name());
			throw new RuntimeException(e);
		}
		
		LongDbQuery query = new LongDbQuery();
		try {
			query.setTableName(mqMessage.getLongMqKey());
			query.setClassz(CnLongMqMessage.class);
			query.setPageSize(1);
			query.addLongDbWhere(LongDbWhere.where(CnLongMqDataSync.longMqId.name()).is(mqMessage.getLongMqId()));
			// 更新已接收的消息数据
			mqMessage.setCompleteDate(new Date());
			mqMessage.setNoticeSenderDate(new Date());
			LongMqServer.getExecuter().getLongDbServer().update(query, mqMessage);
		} catch(Exception e) {
			mqMessage.setErrorCode(CnLongMqExceptionCode.UPDA_MQ_MESSAGE_FAIL_CUSTOMER.name());
			throw new RuntimeException(e);
		}

		try {
			// 异步通知消息的发送方，告知ACK确认结果
			mqMessage.setReqType(CnLongMqDataSync.noticeAckResult.name());
			LongMqRequest.launchSendMessage(mqMessage);
		}
		catch(Exception e) {
			mqMessage.setErrorCode(CnLongMqExceptionCode.NOTICE_ACK_FAIL_CUSTOMER.name());
			throw new RuntimeException(e);
		}
		
		return message;
	}

	/**
	 * 功能描述：消息发送完后，进行消息转移
	 × 模块作者：ZOUYONG
	 × 开发时间：2019年4月17日 下午4:12:37
	 × 返回数据：void
	 */
	public static void processTransferMqMessage(CnLongMqMessage mqMessage, String longMqStatus) {
		LongMqServer.getExecuter().addMqMessageMatching(mqMessage);
		// 开启事务控制
		LongdbTransactionActuator.transactionManager(
			() -> {
				LongDbQuery query = new LongDbQuery();
				query.setTableName(mqMessage.getLongMqKey());
				query.setClassz(CnLongMqMessage.class);
				query.setPageSize(1);
				query.addLongDbWhere(LongDbWhere.where(CnLongMqDataSync.longMqId.name()).is(mqMessage.getLongMqId()));
				
				try {
					// 删除原有数据
					LongMqServer.getExecuter().getLongDbServer().remove(query);
				} catch (Exception e) {
					throw new RuntimeException("ACK确认完成，删除原有消息异常！", e);
				}
				
				try {
					// 插入处理成功消息
					mqMessage.setLongMqStatus(longMqStatus);
					LongMqServer.getExecuter().getLongDbServer().insert(mqMessage, mqMessage.getLongMqKey());
				} catch (Exception e) {
					throw new RuntimeException("ACK确认完成，插入处理成功消息异常！", e);
				}
				return null;
			}
		);
	}

}

